package com.liuyi.netty.mqtt.configution;

import com.liuyi.netty.mqtt.common.helper.RejectHandler;
import com.liuyi.netty.mqtt.common.helper.ThreadFactoryImpl;
import com.liuyi.netty.mqtt.processed.MqttProcessManager;
import com.liuyi.netty.mqtt.processed.process.impl.ConnectMqttProcess;
import com.liuyi.netty.mqtt.processed.process.impl.DisconnectMqttProcess;
import com.liuyi.netty.mqtt.processed.store.MqttSessionStore;
import com.liuyi.netty.mqtt.server.AuthServer;
import io.netty.handler.codec.mqtt.MqttMessageType;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MqttBeanManage {

    private final AuthServer authServer;

    private final MqttSessionStore mqttSessionStore;


    public MqttBeanManage(AuthServer authServer, MqttSessionStore mqttSessionStore) {
        this.authServer = authServer;
        this.mqttSessionStore = mqttSessionStore;
    }

    public void start() {
        int coreThreadNum = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor connectThreadPoolExecutor = new ThreadPoolExecutor(coreThreadNum * 2,
                coreThreadNum * 2,
                60 * 1000,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(100000),
                new ThreadFactoryImpl("connectThreadPoolExecutor"),
                new RejectHandler("connectChannel", 100000));

        ThreadPoolExecutor disconnectThreadPoolExecutor = new ThreadPoolExecutor(coreThreadNum * 2,
                coreThreadNum * 2,
                60 * 1000,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(100000),
                new ThreadFactoryImpl("disconnectThreadPoolExecutor"),
                new RejectHandler("disconnectChannel", 100000));

        MqttProcessManager.putProcess(MqttMessageType.CONNECT, new ConnectMqttProcess(this), connectThreadPoolExecutor);

        MqttProcessManager.putProcess(MqttMessageType.DISCONNECT, new DisconnectMqttProcess(this), disconnectThreadPoolExecutor);
    }

    public AuthServer getAuthServer() {
        return authServer;
    }

    public MqttSessionStore getMqttSessionStore() {
        return mqttSessionStore;
    }
}
